package com.htdata.iiot.tools;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Random;
import java.util.Set;

import org.anarres.lzo.LzoAlgorithm;
import org.anarres.lzo.LzoCompressor;
import org.anarres.lzo.LzoDecompressor;
import org.anarres.lzo.LzoInputStream;
import org.anarres.lzo.LzoLibrary;
import org.anarres.lzo.LzoOutputStream;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.xerial.snappy.Snappy;

import java.util.zip.Deflater;
import java.util.zip.GZIPInputStream;  
import java.util.zip.GZIPOutputStream;  

import com.alibaba.fastjson.JSONObject;

import net.jpountz.lz4.LZ4Compressor;
import net.jpountz.lz4.LZ4Factory;
import net.jpountz.lz4.LZ4FastDecompressor;

public class Post implements Runnable {
	List<ArrayList<Object>> sList = null;
	String clientId = null;

	public MqttEntity mqttEntity;
	public MqttClient mqttClient;
	public Profile  conf = Profile.profile;
	
	public Post(List<ArrayList<Object>> sList, String url, String clientId) {
		this.sList = sList;
		this.mqttEntity = new MqttEntity(clientId);
		try {
			this.mqttClient = mqttEntity.mqttConnect(url);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		} catch (URISyntaxException e) {
			e.printStackTrace();
		}
	}

	void SendPost(int tid, String param) {
		byte[]  data = null;
		String topic = null;
		if(conf.isZlib() == true){
			try {
				data = zlib(param.getBytes());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			topic = String.format( "/hardware/mqtt/%d/zlib", tid);
				try {
				mqttClient.publish(topic, data, 0, false);
			} catch (MqttPersistenceException e) {
				e.printStackTrace();
			} catch (MqttException e) {
				e.printStackTrace();
			}
		}
		if(conf.isGzip() == true){
			try {
				data = gzip(param.getBytes());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			topic = String.format( "/hardware/mqtt/%d/gzip", tid);
				try {
				mqttClient.publish(topic, data, 0, false);
			} catch (MqttPersistenceException e) {
				e.printStackTrace();
			} catch (MqttException e) {
				e.printStackTrace();
			}
		}
		if(conf.isLzo() == true){
			try {
				data = lzo(param.getBytes());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			topic = String.format( "/hardware/mqtt/%d/lzo", tid);
				try {
				mqttClient.publish(topic, data, 0, false);
			} catch (MqttPersistenceException e) {
				e.printStackTrace();
			} catch (MqttException e) {
				e.printStackTrace();
			}
		}
		if(conf.isSnappy() == true){
			try {
				data = snappy(param.getBytes());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			topic = String.format( "/hardware/mqtt/%d/snappy", tid);
				try {
				mqttClient.publish(topic, data, 0, false);
			} catch (MqttPersistenceException e) {
				e.printStackTrace();
			} catch (MqttException e) {
				e.printStackTrace();
			}
			//	System.out.println("size:"+new String(data).toString().length()+"  datasize:"+param.length() );
		}
		if(conf.isLz4() == true){
			try {
				data = lz4(param.getBytes());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			topic = String.format( "/hardware/mqtt/%d/lz4", tid);
			try {
				mqttClient.publish(topic, data, 0, false);
			} catch (MqttPersistenceException e) {
				e.printStackTrace();
			} catch (MqttException e) {
				e.printStackTrace();
			}
		}
		if(conf.isNoCompress()==true){
		
			try {
				mqttClient.publish(String.format( "/hardware/mqtt/%d/json", tid), param.getBytes(), 0, false);
			} catch (MqttPersistenceException e) {
				e.printStackTrace();
			} catch (MqttException e) {
				e.printStackTrace();
			}
		}
	}

	long st = System.currentTimeMillis();

	public void run() {
		int tid = 0;
		while (true) {
			st = System.currentTimeMillis();
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			// 3.生成数据
			for (ArrayList<Object> arr : sList) {
				tid = (int) arr.get(0);
				HashMap<String, Object> hashMap = (HashMap<String, Object>) arr.get(1);
				Set<String> keySet = hashMap.keySet();
				for (String key : keySet) {
					
					if (key.equals("ts")) {
						hashMap.put("ts", st);
					} else if (key.equals("dn")) {
					} else if(key.equals("status")){	
						int statusValue = 0;
						if(hashMap.get(key) instanceof Integer){
							statusValue = Integer.parseInt(hashMap.get(key).toString());
						}
						if(statusValue == 0)
							hashMap.put(key, Math.ceil(new Random().nextDouble() * 10000) * 0.01);
						else
							hashMap.put(key, (new Random().nextInt(4)+ 1) * 1000);

					} else{
						hashMap.put(key, Math.ceil(new Random().nextDouble() * 10000) * 0.01);
					}
				}
				
				if(hashMap.isEmpty())
				{
					System.out.println("hashMap.isEmpty");
				}
				try
				{
					SendPost(tid, buildResult(hashMap));
				}
				catch(Exception e)
				{
					e.printStackTrace();
				}
				
				Auto2Mqtt.mAdder.add(hashMap.size() - 2);
				Auto2Mqtt.adder.increment();
			}
		}
	}
	
	 /**
     * @param srcByte 原始数据
     * @return 压缩后的数据
     */
    public static byte[] lz4(byte[] srcByte) {
        LZ4Factory factory = LZ4Factory.fastestInstance();
        LZ4Compressor compressor = factory.fastCompressor();
        return compressor.compress(srcByte);
    }

    /**
     * @param compressorByte 压缩后的数据
     * @param srcLength      压缩前的数据长度
     * @return
     */
    public static byte[] unLz4(byte[] compressorByte, int srcLength) {
        LZ4Factory factory = LZ4Factory.fastestInstance();
        LZ4FastDecompressor decompressor = factory.fastDecompressor();
      
        return decompressor.decompress(compressorByte, srcLength);
    }


	private static String buildResult(HashMap<String, Object> hashMap) {
		return JSONObject.toJSONString(hashMap).toString();
	}
	
	////	data = Snappy.compress(param.getBytes());
	public static byte[] snappy(byte srcBytes[]) throws IOException {
		return Snappy.compress(srcBytes);
	}
	
	public static byte[] lzo(byte srcBytes[]) throws IOException {
        LzoCompressor compressor = LzoLibrary.getInstance().newCompressor(
                LzoAlgorithm.LZO1X, null);

        ByteArrayOutputStream os = new ByteArrayOutputStream();
        LzoOutputStream cs = new LzoOutputStream(os, compressor);
        cs.write(srcBytes);
        cs.close();

        return os.toByteArray();
    }


    public static byte[] gzip(byte[] data) throws Exception {  
        ByteArrayOutputStream bos = new ByteArrayOutputStream();  
        GZIPOutputStream gzip = new GZIPOutputStream(bos);  
        gzip.write(data);  
        gzip.finish();  
        gzip.close();  
        byte[] ret = bos.toByteArray();  
        bos.close();  
        return ret;  
    } 
    
    public static byte[] zlib(byte[] data) {  
        byte[] output = new byte[0];  
  
        Deflater compresser = new Deflater();  
  
        compresser.reset();  
        compresser.setInput(data);  
        compresser.finish();  
        ByteArrayOutputStream bos = new ByteArrayOutputStream(data.length);  
        try {  
            byte[] buf = new byte[1024];  
            while (!compresser.finished()) {  
                int i = compresser.deflate(buf);  
                bos.write(buf, 0, i);  
            }  
            output = bos.toByteArray();  
        } catch (Exception e) {  
            output = data;  
            e.printStackTrace();  
        } finally {  
            try {  
                bos.close();  
            } catch (IOException e) {  
                e.printStackTrace();  
            }  
        }  
        compresser.end();  
        return output;  
    }  
	
}
